package com.atguigu.gmall.realtime.function;


import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.util.DimUtil;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import com.atguigu.gmall.realtime.util.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author lzc
 * @Date 2023/2/21 14:31
 */
public abstract class AsyncDimFunction<T> extends RichAsyncFunction<T, T> {
    
    private ThreadPoolExecutor threadPool;
    
    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
    }
    
    public abstract String getTable();
    
    public abstract Object getId(T value);
    
    public abstract void addDim(T value, JSONObject dim);
    
    @Override
    public void asyncInvoke(T input,
                            ResultFuture<T> resultFuture) throws Exception {
        // 每来一条数据, 都提交一个线程去补充维度信息
        threadPool.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    // 获取redis 客户端
                    Jedis redisClient = RedisUtil.getRedisClient();
                    // 获取 phoenix 客户端
                    DruidPooledConnection phoenixConn = DruidDSUtil.getPhoenixConn();
                    JSONObject dim = DimUtil.readDim(redisClient, phoenixConn, getTable(), getId(input));
                    addDim(input, dim);
                    // 归还资源给连接池
                    redisClient.close();
                    phoenixConn.close();
                    
                    // 把补充完维度的 input 对象, 放入到resultFuture
                    resultFuture.complete(Collections.singletonList(input));
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
    
                // 读取维度数据
                
                // 补充
                
                // 输出数据
                
                
            }
        });
    }
    
    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        throw new RuntimeException(" 无法读取相应的维度: " + input + " 比较检测集群, 检测数据是否全....");
    }
}
